ποΈGitΠ―ΡΠ°ποΈ
Commit 4a9f85bf5e54ca8b5089837db2a6b9382d77875f
Parents : 2fa0abd
Author : James Rich <2199651+jamesarich@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-07-05T20:41:29-05:00
Committer : GitHub <noreply@github.com>
Date : 2026-07-06T01:41:29Z
feat(database): unify device database across transports (#6096)
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
Changes
16 files changed, 558 insertions(+), 6 deletions(-)
Diff
diff --git a/core/common/src/commonMain/kotlin/org/meshtastic/core/common/database/DatabaseManager.kt b/core/common/src/commonMain/kotlin/org/meshtastic/core/common/database/DatabaseManager.kt
index 9e84b74ee0..adf0b93dc4 100644
--- a/core/common/src/commonMain/kotlin/org/meshtastic/core/common/database/DatabaseManager.kt
+++ b/core/common/src/commonMain/kotlin/org/meshtastic/core/common/database/DatabaseManager.kt
@@ -32,6 +32,14 @@ interface DatabaseManager {
/** Switches the active database to the one associated with the given [address]. */
suspend fun switchActiveDatabase(address: String?)
+ /**
+ * Associates the currently-active connection with the physical node [nodeNum] learned from the radio handshake, so
+ * the same node reached over a different transport (BLE / TCP / USB) shares one database. The first transport to
+ * report a node claims the active DB for it; a later transport folds its own data in and switches to the claimed
+ * DB. Idempotent once a node/transport pair has been unified.
+ */
+ suspend fun associateNode(nodeNum: Int)
+
/** Returns true if a database exists for the given device address. */
fun hasDatabaseFor(address: String?): Boolean
}
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseConstants.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseConstants.kt
index d72187b9b7..589bceb774 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseConstants.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseConstants.kt
@@ -32,6 +32,11 @@ object DatabaseConstants {
const val LEGACY_DB_CLEANED_KEY: String = "legacy_db_cleaned"
+ // Cross-transport unification: map a node / transport address to the DB file that node's data lives in.
+ // Keys are dynamic (suffixed with the node num / normalized address), mirroring the `db_last_used:` pattern.
+ const val NODE_DB_FOR_PREFIX: String = "node_db_for:"
+ const val ADDR_DB_FOR_PREFIX: String = "addr_db_for:"
+
// Display/truncation and hash sizing for DB names
const val DB_NAME_HASH_LEN: Int = 10
const val DB_NAME_SEPARATOR_LEN: Int = 1
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt
index 3511c4e598..bbfe75cbee 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt
@@ -22,6 +22,7 @@ import androidx.datastore.preferences.core.booleanPreferencesKey
import androidx.datastore.preferences.core.edit
import androidx.datastore.preferences.core.intPreferencesKey
import androidx.datastore.preferences.core.longPreferencesKey
+import androidx.datastore.preferences.core.stringPreferencesKey
import co.touchlab.kermit.Logger
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
@@ -44,6 +45,7 @@ import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext
import org.koin.core.annotation.Named
import org.koin.core.annotation.Single
+import org.meshtastic.core.common.util.normalizeAddress
import org.meshtastic.core.common.util.nowMillis
import org.meshtastic.core.di.CoroutineDispatchers
import kotlin.concurrent.Volatile
@@ -67,6 +69,11 @@ open class DatabaseManager(
private fun lastUsedKey(dbName: String) = longPreferencesKey("db_last_used:$dbName")
+ private fun addrDbKey(address: String?) =
+ stringPreferencesKey("${DatabaseConstants.ADDR_DB_FOR_PREFIX}${normalizeAddress(address)}")
+
+ private fun nodeDbKey(nodeNum: Int) = stringPreferencesKey("${DatabaseConstants.NODE_DB_FOR_PREFIX}$nodeNum")
+
private var backfillJob: Job? = null
@Volatile private var hasDelayedFirstDeviceBackfill = false
@@ -83,9 +90,7 @@ open class DatabaseManager(
managerScope.launch {
datastore.edit { it[cacheLimitKey] = clamped }
// Enforce asynchronously with current active DB protected
- val active =
- _currentDb.value?.let { buildDbName(_currentAddress.value) } ?: DatabaseConstants.DEFAULT_DB_NAME
- enforceCacheLimit(activeDbName = active)
+ enforceCacheLimit(activeDbName = currentDbName)
}
}
@@ -105,6 +110,13 @@ open class DatabaseManager(
private val _currentAddress = MutableStateFlow<String?>(null)
val currentAddress: StateFlow<String?> = _currentAddress
+ /**
+ * Name of the currently active database. Tracked explicitly rather than recomputed from the address, because
+ * cross-transport aliasing ([associateNode]) decouples the two: a secondary transport's address maps to the DB
+ * claimed by the first transport, which `buildDbName(address)` would never produce. Written under [mutex].
+ */
+ @Volatile private var currentDbName: String = DatabaseConstants.DEFAULT_DB_NAME
+
/** Initialize the active database for [address]. */
suspend fun init(address: String?) {
switchActiveDatabase(address)
@@ -118,12 +130,24 @@ open class DatabaseManager(
private fun getOrOpenDatabase(dbName: String): MeshtasticDatabase =
dbCache.getOrPut(dbName) { getDatabaseBuilder(dbName).build() }
+ /**
+ * Resolves the DB name to use for [address], honoring a cross-transport alias when one exists. A secondary
+ * transport (e.g. TCP) that has been unified with a node points at the DB the first transport (e.g. BLE) claimed;
+ * without an alias this falls back to the address-hashed name β today's default β for a first-time or primary
+ * connection. See [associateNode].
+ */
+ private suspend fun resolveDbName(address: String?): String {
+ val fallback = buildDbName(address)
+ if (fallback == DatabaseConstants.DEFAULT_DB_NAME) return fallback
+ return datastore.data.first()[addrDbKey(address)] ?: fallback
+ }
+
/** Switch active database to the one associated with [address]. Serialized via mutex. */
override suspend fun switchActiveDatabase(address: String?) = mutex.withLock {
- val dbName = buildDbName(address)
+ val dbName = resolveDbName(address)
// Remember the previously active DB name (any) so we can record its last-used time as well.
- val previousDbName = _currentDb.value?.let { buildDbName(_currentAddress.value) }
+ val previousDbName = if (_currentDb.value != null) currentDbName else null
// Fast path: no-op if already on this address
if (_currentAddress.value == address && _currentDb.value != null) {
@@ -140,6 +164,7 @@ open class DatabaseManager(
// collectors, causing "Connection pool is closed" crashes.
_currentDb.value = db
_currentAddress.value = address
+ currentDbName = dbName
markLastUsed(dbName)
// Also mark the previous DB as used "just now" so LRU has an accurate, recent timestamp
previousDbName?.let { markLastUsed(it) }
@@ -167,6 +192,78 @@ open class DatabaseManager(
Logger.i { "Switched active DB to ${anonymizeDbName(dbName)} for address ${anonymizeAddress(address)}" }
}
+ @Suppress("TooGenericExceptionCaught")
+ override suspend fun associateNode(nodeNum: Int) {
+ mutex.withLock {
+ val sourceName = currentDbName
+ // Never claim or merge into the sentinel "no device" DB.
+ if (sourceName == DatabaseConstants.DEFAULT_DB_NAME) return@withLock
+
+ val claimed = datastore.data.first()[nodeDbKey(nodeNum)]
+ when {
+ claimed == null -> {
+ // First transport to learn this node: its current DB becomes the node's canonical DB.
+ // No address alias is needed β a primary connection already resolves to this DB via buildDbName.
+ datastore.edit { it[nodeDbKey(nodeNum)] = sourceName }
+ Logger.i { "Claimed ${anonymizeDbName(sourceName)} as canonical DB for node $nodeNum" }
+ }
+
+ claimed == sourceName -> Unit
+
+ // Already unified β nothing to do.
+
+ else -> {
+ // Secondary transport reached an already-known node: fold this DB into the canonical one,
+ // switch the active DB to it, alias this address to it, and retire the now-merged source.
+ val source = _currentDb.value ?: return@withLock
+ val dest = withContext(dispatchers.io) { getOrOpenDatabase(claimed) }
+
+ // Redirect live writers to the canonical DB BEFORE merging. Otherwise a concurrent withDb
+ // write (connect triggers a full NodeDB re-dump) could land in `source` after the merge has
+ // already snapshotted it, then be lost when `source` is retired β and withDb's closed-pool
+ // retry can't recover it because the write itself succeeded. New writers now capture `dest`;
+ // any still holding `source` are covered by that retry once retireDatabase closes it.
+ _currentDb.value = dest
+ currentDbName = claimed
+ try {
+ withContext(dispatchers.io) { DatabaseMerger.merge(source, dest) }
+ } catch (e: CancellationException) {
+ throw e
+ } catch (e: Exception) {
+ // merge() is atomic, so on failure `dest` is unchanged. Roll the active DB back to
+ // `source` so the address still resolves consistently and the merge retries next connect.
+ _currentDb.value = source
+ currentDbName = sourceName
+ Logger.w(e) {
+ "Merge into ${anonymizeDbName(claimed)} failed; kept ${anonymizeDbName(sourceName)} active"
+ }
+ return@withLock
+ }
+
+ markLastUsed(claimed)
+ datastore.edit { it[addrDbKey(_currentAddress.value)] = claimed }
+ Logger.i {
+ "Unified ${anonymizeDbName(sourceName)} into ${anonymizeDbName(claimed)} for node $nodeNum"
+ }
+
+ // Retire the merged source off the critical path; its data now lives in the canonical DB.
+ managerScope.launch(dispatchers.io) { retireDatabase(sourceName) }
+ }
+ }
+ }
+ }
+
+ /** Closes, deletes, and forgets a database whose contents have been merged into another. */
+ private suspend fun retireDatabase(dbName: String) = mutex.withLock {
+ runCatching {
+ closeCachedDatabase(dbName)
+ deleteDatabase(dbName)
+ datastore.edit { it.remove(lastUsedKey(dbName)) }
+ }
+ .onSuccess { Logger.i { "Retired merged DB ${anonymizeDbName(dbName)}" } }
+ .onFailure { Logger.w(it) { "Failed to retire merged database ${anonymizeDbName(dbName)}" } }
+ }
+
/**
* Closes and removes a cached database by name. Safe to call even if the database was already closed or not in the
* cache. Does NOT delete the underlying file β the database can be re-opened on next access.
@@ -188,7 +285,7 @@ open class DatabaseManager(
@Suppress("TooGenericExceptionCaught")
override suspend fun <T> withDb(block: suspend (MeshtasticDatabase) -> T): T? = withContext(limitedIo) {
val db = _currentDb.value ?: return@withContext null
- val active = buildDbName(_currentAddress.value)
+ val active = currentDbName
markLastUsed(active)
try {
block(db)
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseMerger.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseMerger.kt
new file mode 100644
index 0000000000..3dfda63840
--- /dev/null
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseMerger.kt
@@ -0,0 +1,148 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.database
+
+import androidx.room3.immediateTransaction
+import androidx.room3.useWriterConnection
+import co.touchlab.kermit.Logger
+
+/**
+ * Folds the contents of one device database ([source]) into another ([dest]) when both turn out to belong to the same
+ * physical node reached over different transports (BLE / TCP / USB). Called once, by [DatabaseManager.associateNode],
+ * the first time a secondary transport learns a `myNodeNum` that another transport already claimed.
+ *
+ * Every per-device table is unified so switching transport is seamless: messages (+FTS), reactions, contact mute/read
+ * settings, nodes and their notes, per-node metadata, the audit log (each session's received-packet history β the
+ * source of the telemetry timelines, position history, and traceroute results the UI reconstructs), traceroute
+ * positions, and discovery sessions. Where the same row exists on both sides the destination is preferred (it will be
+ * refreshed by the same radio's re-dump on connect); source-only rows and strictly newer history are brought over.
+ * Packets/reactions in [source] and [dest] already share the same `myNodeNum` (same node), so no key remapping is
+ * needed there; autoincrement-keyed rows (packets, discovery) are re-inserted with fresh ids to avoid collisions.
+ */
+object DatabaseMerger {
+
+ suspend fun merge(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+ // All destination writes run in a single transaction so a crash or exception mid-merge rolls
+ // back cleanly instead of leaving `dest` half-merged. This also makes a retried merge safe: if
+ // associateNode re-runs (e.g. a crash before the address alias is persisted), the rolled-back
+ // partial writes never happened, so the fresh-id packet/discovery re-inserts can't duplicate.
+ // Reads from `source` use its own connection pool and don't participate in this transaction.
+ var packets = 0
+ dest.useWriterConnection { transactor ->
+ transactor.immediateTransaction {
+ packets = mergePackets(source, dest)
+ mergeReactions(source, dest)
+ mergeContactSettings(source, dest)
+ mergeNodes(source, dest)
+ mergeMetadata(source, dest)
+ // Logs must precede traceroute positions: the latter has a CASCADE foreign key onto log.uuid.
+ mergeLogs(source, dest)
+ mergeTraceroutePositions(source, dest)
+ mergeDiscovery(source, dest)
+ }
+ }
+ Logger.i { "Merged $packets packets across transports into unified node DB" }
+ }
+
+ /**
+ * Message history is the primary payload. `uuid` is per-DB autoincrement, so re-insert with uuid = 0 to mint a
+ * fresh id and avoid collisions; rebuild the FTS index once at the end so copied messages are searchable.
+ */
+ private suspend fun mergePackets(source: MeshtasticDatabase, dest: MeshtasticDatabase): Int {
+ val packets = source.packetDao().getAllPacketsSnapshot()
+ packets.forEach { dest.packetDao().insertPacketForMerge(it.copy(uuid = 0L)) }
+ if (packets.isNotEmpty()) dest.packetDao().rebuildFtsIndex()
+ return packets.size
+ }
+
+ /** Composite PK (myNodeNum, reply_id, user_id, emoji) makes IGNORE a natural dedupe. */
+ private suspend fun mergeReactions(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+ dest.packetDao().insertReactionsIgnore(source.packetDao().getAllReactionsSnapshot())
+ }
+
+ /** Keep the destination's settings on conflict (IGNORE); only bring over contacts it has never seen. */
+ private suspend fun mergeContactSettings(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+ dest.packetDao().insertContactSettingsIgnore(source.packetDao().getAllContactSettingsSnapshot())
+ }
+
+ /**
+ * Bring over every node the destination has never seen (so its last-known identity/telemetry/position shows
+ * immediately, including offline nodes the radio may not re-report). For overlapping nodes, keep the destination's
+ * row β the same radio re-dumps it on connect β but fill in a user note where the destination has none, never
+ * clobbering one the user already wrote on the canonical DB.
+ */
+ private suspend fun mergeNodes(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+ val destNums = dest.nodeInfoDao().getAllNodesSnapshot().mapTo(mutableSetOf()) { it.num }
+ source.nodeInfoDao().getAllNodesSnapshot().forEach { node ->
+ when {
+ node.num !in destNums -> dest.nodeInfoDao().upsert(node)
+
+ node.notes.isNotBlank() -> {
+ val existing = dest.nodeInfoDao().getNodeByNum(node.num)
+ if (existing != null && existing.node.notes.isBlank()) {
+ dest.nodeInfoDao().setNodeNotes(node.num, node.notes)
+ }
+ }
+ }
+ }
+ }
+
+ /** Per-node DeviceMetadata: bring source rows the dest lacks or that are strictly newer (by timestamp). */
+ private suspend fun mergeMetadata(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+ val destByNum = dest.nodeInfoDao().getAllMetadataSnapshot().associateBy { it.num }
+ source.nodeInfoDao().getAllMetadataSnapshot().forEach { meta ->
+ val existing = destByNum[meta.num]
+ if (existing == null || meta.timestamp > existing.timestamp) dest.nodeInfoDao().upsert(meta)
+ }
+ }
+
+ /**
+ * The audit log holds each transport session's received-packet history β the source of the telemetry timelines,
+ * position history, and node-info history the UI reconstructs per node/port. This is the only place that time
+ * series diverges between transports (the current node snapshot is re-dumped identically by the same radio), so
+ * merging it is what actually unifies "metrics/telemetry/positions". uuid is a unique string, so IGNORE is safe.
+ */
+ private suspend fun mergeLogs(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+ dest.meshLogDao().insertIgnore(source.meshLogDao().getAllLogsSnapshot())
+ }
+
+ /**
+ * Traceroute-derived positions, keyed to their log entry (CASCADE FK) β merge after [mergeLogs] so the parent log
+ * rows already exist in [dest].
+ */
+ private suspend fun mergeTraceroutePositions(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+ dest.tracerouteNodePositionDao().insertAll(source.tracerouteNodePositionDao().getAllSnapshot())
+ }
+
+ /**
+ * Discovery sessions are distinct events per transport session, so append them all (no dedupe). The three tables
+ * form an autoincrement-keyed FK chain (session β preset result β discovered node), so re-insert each level with a
+ * fresh id and rewire the child's foreign key to the parent's newly-assigned id.
+ */
+ private suspend fun mergeDiscovery(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+ val src = source.discoveryDao()
+ val dst = dest.discoveryDao()
+ src.getAllSessionsSnapshot().forEach { session ->
+ val newSessionId = dst.insertSession(session.copy(id = 0))
+ src.getPresetResults(session.id).forEach { preset ->
+ val newPresetId = dst.insertPresetResult(preset.copy(id = 0, sessionId = newSessionId))
+ val nodes = src.getDiscoveredNodes(preset.id).map { it.copy(id = 0, presetResultId = newPresetId) }
+ if (nodes.isNotEmpty()) dst.insertDiscoveredNodes(nodes)
+ }
+ }
+ }
+}
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt
index 7e4ebdf5ed..4a115fbea8 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt
@@ -39,6 +39,10 @@ interface DiscoveryDao {
@Query("SELECT * FROM discovery_session ORDER BY timestamp DESC")
fun getAllSessions(): Flow<List<DiscoverySessionEntity>>
+ /** Snapshot used by DatabaseMerger to append a transport's discovery sessions to the unified DB. */
+ @Query("SELECT * FROM discovery_session")
+ suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity>
+
@Query("SELECT * FROM discovery_session WHERE id = :sessionId")
suspend fun getSession(sessionId: Long): DiscoverySessionEntity?
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/MeshLogDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/MeshLogDao.kt
index 3edabfc150..9369bbf0e7 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/MeshLogDao.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/MeshLogDao.kt
@@ -18,6 +18,7 @@ package org.meshtastic.core.database.dao
import androidx.room3.Dao
import androidx.room3.Insert
+import androidx.room3.OnConflictStrategy
import androidx.room3.Query
import kotlinx.coroutines.flow.Flow
import org.meshtastic.core.database.entity.MeshLog
@@ -47,6 +48,16 @@ interface MeshLogDao {
@Insert suspend fun insert(log: MeshLog)
+ /**
+ * Snapshot + IGNORE-insert used by DatabaseMerger to carry telemetry/position/traceroute history across transports.
+ * uuid is a unique string per row, so IGNORE is just belt-and-suspenders against a rare clash.
+ */
+ @Query("SELECT * FROM log")
+ suspend fun getAllLogsSnapshot(): List<MeshLog>
+
+ @Insert(onConflict = OnConflictStrategy.IGNORE)
+ suspend fun insertIgnore(logs: List<MeshLog>)
+
@Query("DELETE FROM log")
suspend fun deleteAll()
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/NodeInfoDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/NodeInfoDao.kt
index 2966e4e49c..39686488ec 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/NodeInfoDao.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/NodeInfoDao.kt
@@ -283,6 +283,10 @@ interface NodeInfoDao {
@Query("DELETE FROM metadata WHERE num=:num")
suspend fun deleteMetadata(num: Int)
+ /** Snapshot used by DatabaseMerger to carry per-node DeviceMetadata across transports (newest timestamp wins). */
+ @Query("SELECT * FROM metadata")
+ suspend fun getAllMetadataSnapshot(): List<MetadataEntity>
+
@Query("SELECT * FROM nodes WHERE num=:num")
@Transaction
suspend fun getNodeByNum(num: Int): NodeWithRelations?
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt
index a7083e37d0..801f20b86f 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt
@@ -489,6 +489,26 @@ interface PacketDao {
@Query("DELETE FROM contact_settings")
suspend fun deleteAllContactSettings()
+ // region ββ Cross-transport merge ββ
+ // Snapshots + inserts used by DatabaseMerger to fold one transport's DB into another for the same node.
+
+ @Query("SELECT * FROM packet")
+ suspend fun getAllPacketsSnapshot(): List<Packet>
+
+ /** Insert a packet copied from another DB. Pass uuid = 0 so a fresh auto-generated id is assigned. */
+ @Insert suspend fun insertPacketForMerge(packet: Packet)
+
+ @Query("SELECT * FROM reactions")
+ suspend fun getAllReactionsSnapshot(): List<ReactionEntity>
+
+ @Insert(onConflict = OnConflictStrategy.IGNORE)
+ suspend fun insertReactionsIgnore(reactions: List<ReactionEntity>)
+
+ @Query("SELECT * FROM contact_settings")
+ suspend fun getAllContactSettingsSnapshot(): List<ContactSettings>
+
+ // endregion
+
/**
* One-time migration: Remap all message DataPacket.channel indices to new mapping using PSK after a channel
* reorder. For each Packet (with port_num = 1), finds the old PSK then sets the channel index to the matching
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/TracerouteNodePositionDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/TracerouteNodePositionDao.kt
index ffd0ecc2b3..85988f781b 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/TracerouteNodePositionDao.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/TracerouteNodePositionDao.kt
@@ -31,5 +31,8 @@ interface TracerouteNodePositionDao {
@Query("DELETE FROM traceroute_node_position WHERE log_uuid = :logUuid")
suspend fun deleteByLogUuid(logUuid: String)
+ @Query("SELECT * FROM traceroute_node_position")
+ suspend fun getAllSnapshot(): List<TracerouteNodePositionEntity>
+
@Upsert suspend fun insertAll(entities: List<TracerouteNodePositionEntity>)
}
diff --git a/core/database/src/commonTest/kotlin/org/meshtastic/core/database/DatabaseMergerTest.kt b/core/database/src/commonTest/kotlin/org/meshtastic/core/database/DatabaseMergerTest.kt
new file mode 100644
index 0000000000..e91ac184f1
--- /dev/null
+++ b/core/database/src/commonTest/kotlin/org/meshtastic/core/database/DatabaseMergerTest.kt
@@ -0,0 +1,223 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.database
+
+import kotlinx.coroutines.test.runTest
+import okio.ByteString.Companion.toByteString
+import org.meshtastic.core.database.entity.ContactSettings
+import org.meshtastic.core.database.entity.DiscoveredNodeEntity
+import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
+import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.MeshLog
+import org.meshtastic.core.database.entity.MetadataEntity
+import org.meshtastic.core.database.entity.MyNodeEntity
+import org.meshtastic.core.database.entity.NodeEntity
+import org.meshtastic.core.database.entity.Packet
+import org.meshtastic.core.database.entity.ReactionEntity
+import org.meshtastic.core.database.entity.TracerouteNodePositionEntity
+import org.meshtastic.core.model.DataPacket
+import org.meshtastic.core.model.NodeAddress
+import org.meshtastic.proto.DeviceMetadata
+import org.meshtastic.proto.HardwareModel
+import org.meshtastic.proto.PortNum
+import org.meshtastic.proto.Position
+import org.meshtastic.proto.User
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+/**
+ * Verifies [DatabaseMerger] folds a secondary transport's DB into the canonical one for the same node: message history
+ * combines with fresh non-colliding ids and stays searchable, node notes are preserved without clobbering, and
+ * reactions dedupe on their composite key. This is the data-integrity path for cross-transport DB unification.
+ */
+class DatabaseMergerTest {
+
+ // Same physical node reached over two transports β same myNodeNum in both DBs.
+ private val nodeNum = 0x1234
+
+ private val source: MeshtasticDatabase = getInMemoryDatabaseBuilder().build()
+ private val dest: MeshtasticDatabase = getInMemoryDatabaseBuilder().build()
+
+ @AfterTest
+ fun tearDown() {
+ source.close()
+ dest.close()
+ }
+
+ private fun myNode() = MyNodeEntity(
+ myNodeNum = nodeNum,
+ model = "TBEAM",
+ firmwareVersion = "2.5.0",
+ couldUpdate = false,
+ shouldUpdate = false,
+ currentPacketId = 1L,
+ messageTimeoutMsec = 300_000,
+ minAppVersion = 1,
+ maxChannels = 8,
+ hasWifi = false,
+ )
+
+ private fun textPacket(contact: String, text: String, time: Long) = Packet(
+ uuid = 0L, // auto-generated
+ myNodeNum = nodeNum,
+ port_num = PortNum.TEXT_MESSAGE_APP.value,
+ contact_key = contact,
+ received_time = time,
+ read = true,
+ data =
+ DataPacket(
+ to = NodeAddress.ID_BROADCAST,
+ bytes = text.encodeToByteArray().toByteString(),
+ dataType = PortNum.TEXT_MESSAGE_APP.value,
+ ),
+ messageText = text, // FTS indexes this column
+ )
+
+ private fun node(num: Int, notes: String = "") = NodeEntity(
+ num = num,
+ user = User(id = "!$num", long_name = "Node $num", hw_model = HardwareModel.TBEAM),
+ notes = notes,
+ )
+
+ private fun logEntry(uuid: String, portNum: Int, time: Long) = MeshLog(
+ uuid = uuid,
+ message_type = "Packet",
+ received_date = time,
+ raw_message = "",
+ fromNum = 30,
+ portNum = portNum,
+ )
+
+ @Test
+ fun mergeFoldsSecondaryTransportIntoCanonical() = runTest {
+ // Canonical (dest): one message, a noted node (#10), an un-noted node (#20), a contact setting, a reaction.
+ dest.nodeInfoDao().setMyNodeInfo(myNode())
+ dest.packetDao().insert(textPacket("0!broadcast", "destmsg", time = 100))
+ dest.nodeInfoDao().upsert(node(10, notes = "keep me"))
+ dest.nodeInfoDao().upsert(node(20)) // no notes yet
+ dest.packetDao().upsertContactSettings(listOf(ContactSettings(contact_key = "0!broadcast", muteUntil = 999)))
+ dest
+ .packetDao()
+ .insert(ReactionEntity(myNodeNum = nodeNum, replyId = 1, userId = "!u", emoji = "π", timestamp = 1))
+ dest.meshLogDao().insert(logEntry("destlog", PortNum.TELEMETRY_APP.value, time = 50))
+ dest
+ .nodeInfoDao()
+ .upsert(MetadataEntity(num = 10, proto = DeviceMetadata(), timestamp = 1000)) // newer than src
+
+ // Secondary (source): two messages, notes for #20 (dest blank β fill) and a source-only node (#30 β insert),
+ // the same reaction (dedupe) plus a new one, and a conflicting contact setting (dest's must win).
+ source.nodeInfoDao().setMyNodeInfo(myNode())
+ source.packetDao().insert(textPacket("0!broadcast", "srcmsgone", time = 200))
+ source.packetDao().insert(textPacket("0!broadcast", "srcmsgtwo", time = 300))
+ source.nodeInfoDao().upsert(node(20, notes = "src note"))
+ source.nodeInfoDao().upsert(node(30, notes = "src-only note"))
+ source.packetDao().upsertContactSettings(listOf(ContactSettings(contact_key = "0!broadcast", muteUntil = 1)))
+ source
+ .packetDao()
+ .insert(ReactionEntity(myNodeNum = nodeNum, replyId = 1, userId = "!u", emoji = "π", timestamp = 1))
+ source
+ .packetDao()
+ .insert(ReactionEntity(myNodeNum = nodeNum, replyId = 1, userId = "!v", emoji = "β€οΈ", timestamp = 2))
+ // Telemetry + position history (audit log) plus a traceroute position referencing a log entry.
+ source.meshLogDao().insert(logEntry("srctelemetry", PortNum.TELEMETRY_APP.value, time = 200))
+ source.meshLogDao().insert(logEntry("srcposition", PortNum.POSITION_APP.value, time = 210))
+ source
+ .tracerouteNodePositionDao()
+ .insertAll(
+ listOf(
+ TracerouteNodePositionEntity(
+ logUuid = "srcposition",
+ requestId = 7,
+ nodeNum = 30,
+ position = Position(),
+ ),
+ ),
+ )
+ source.nodeInfoDao().upsert(node(40)) // source-only node, no notes β still brought over
+ source
+ .nodeInfoDao()
+ .upsert(MetadataEntity(num = 10, proto = DeviceMetadata(), timestamp = 500)) // older β dest wins
+ source
+ .nodeInfoDao()
+ .upsert(MetadataEntity(num = 30, proto = DeviceMetadata(), timestamp = 500)) // dest lacks β added
+ val srcSession =
+ source
+ .discoveryDao()
+ .insertSession(
+ DiscoverySessionEntity(timestamp = 1, presetsScanned = "LongFast", homePreset = "LongFast"),
+ )
+ val srcPreset =
+ source
+ .discoveryDao()
+ .insertPresetResult(DiscoveryPresetResultEntity(sessionId = srcSession, presetName = "LongFast"))
+ source
+ .discoveryDao()
+ .insertDiscoveredNodes(listOf(DiscoveredNodeEntity(presetResultId = srcPreset, nodeNum = 30L)))
+
+ DatabaseMerger.merge(source, dest)
+
+ // Packets: 1 + 2, with fresh distinct non-zero uuids (no collision with dest's existing uuid).
+ val packets = dest.packetDao().getAllPacketsSnapshot()
+ assertEquals(3, packets.size, "all messages combined")
+ assertEquals(3, packets.map { it.uuid }.toSet().size, "uuids are unique")
+ assertTrue(packets.none { it.uuid == 0L }, "every merged packet got a real id")
+
+ // Merged messages remain searchable via the rebuilt FTS index.
+ assertEquals(1, dest.packetDao().searchMessages("srcmsgone").size, "copied message is searchable")
+
+ // Node notes: dest's existing note kept, blank note filled from source, source-only node brought over.
+ assertEquals("keep me", dest.nodeInfoDao().getNodeByNum(10)?.node?.notes)
+ assertEquals("src note", dest.nodeInfoDao().getNodeByNum(20)?.node?.notes)
+ assertNotNull(dest.nodeInfoDao().getNodeByNum(30), "source-only noted node inserted")
+
+ // Reactions dedupe on composite PK: R1 (already in dest) + R2 (new) = 2.
+ assertEquals(2, dest.packetDao().getAllReactionsSnapshot().size)
+
+ // Contact settings: destination's value wins on conflict (IGNORE), never overwritten.
+ assertEquals(
+ 999,
+ dest.packetDao().getAllContactSettingsSnapshot().first { it.contact_key == "0!broadcast" }.muteUntil,
+ )
+
+ // Telemetry/position history lives in the audit log: both transports' entries combine, and traceroute
+ // positions come along with their (now-present) parent log rows.
+ assertEquals(3, dest.meshLogDao().getAllLogsSnapshot().size, "log history from both transports combined")
+ assertEquals(1, dest.tracerouteNodePositionDao().getAllSnapshot().size, "traceroute positions merged")
+
+ // Source-only nodes are brought over even without notes.
+ assertNotNull(dest.nodeInfoDao().getNodeByNum(40), "source-only node brought over")
+
+ // Metadata: newest timestamp wins on overlap, source-only rows added.
+ val metadata = dest.nodeInfoDao().getAllMetadataSnapshot().associateBy { it.num }
+ assertEquals(1000, metadata[10]?.timestamp, "dest's newer metadata kept")
+ assertNotNull(metadata[30], "source-only metadata inserted")
+
+ // Discovery: the session β preset β node chain is appended with rewired foreign keys.
+ val sessions = dest.discoveryDao().getAllSessionsSnapshot()
+ assertEquals(1, sessions.size, "discovery session appended")
+ val presets = dest.discoveryDao().getPresetResults(sessions.first().id)
+ assertEquals(1, presets.size, "preset appended under rewired session id")
+ assertEquals(
+ 1,
+ dest.discoveryDao().getDiscoveredNodes(presets.first().id).size,
+ "node appended under rewired preset id",
+ )
+ }
+}
diff --git a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt
index 0daded9f3f..d52e93c09b 100644
--- a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt
+++ b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt
@@ -19,6 +19,9 @@ package org.meshtastic.core.service
import co.touchlab.kermit.Logger
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.launch
import org.meshtastic.core.common.database.DatabaseManager
import org.meshtastic.core.model.ConnectionState
import org.meshtastic.core.repository.AdminController
@@ -87,6 +90,18 @@ class RadioControllerImpl(
NodeController by NodeControllerImpl(commandSender, nodeManager, packetRepository, scope),
QueryController by QueryControllerImpl(commandSender, nodeManager, uiPrefs) {
+ init {
+ // Unify per-node databases across transports. When the handshake reports our node number, tell the
+ // DatabaseManager to claim (or merge into) that node's canonical DB, so the same node reached over BLE, TCP,
+ // or USB shares one stored history. Keyed on (address, node) so a second transport for the same node re-fires
+ // even though the node number itself is unchanged.
+ scope.launch {
+ combine(meshPrefs.deviceAddress, nodeManager.myNodeNum) { address, nodeNum -> address to nodeNum }
+ .distinctUntilChanged()
+ .collect { (_, nodeNum) -> nodeNum?.let { databaseManager.associateNode(it) } }
+ }
+ }
+
// ββ Connection State ββββββββββββββββββββββββββββββββββββββββββββββββββββ
override val connectionState: StateFlow<ConnectionState>
diff --git a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeDatabaseManager.kt b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeDatabaseManager.kt
index 5693f112f7..5929cebef7 100644
--- a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeDatabaseManager.kt
+++ b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeDatabaseManager.kt
@@ -29,12 +29,14 @@ class FakeDatabaseManager :
override val cacheLimit: StateFlow<Int> = _cacheLimit
var lastSwitchedAddress: String? = null
+ var lastAssociatedNode: Int? = null
val existingDatabases = mutableSetOf<String>()
init {
registerResetAction {
_cacheLimit.value = DEFAULT_CACHE_LIMIT
lastSwitchedAddress = null
+ lastAssociatedNode = null
existingDatabases.clear()
}
}
@@ -49,6 +51,10 @@ class FakeDatabaseManager :
lastSwitchedAddress = address
}
+ override suspend fun associateNode(nodeNum: Int) {
+ lastAssociatedNode = nodeNum
+ }
+
override fun hasDatabaseFor(address: String?): Boolean = address != null && existingDatabases.contains(address)
companion object {
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt
index d4558cd061..0ba7584658 100644
--- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt
@@ -185,6 +185,8 @@ private class HistoryTestDao : DiscoveryDao {
override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> = sessionsFlow
+ override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
+
override suspend fun getSession(sessionId: Long) = sessions[sessionId]
override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt
index 6a62fced89..b334fda62f 100644
--- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt
@@ -171,6 +171,8 @@ private class MapTestDao : DiscoveryDao {
override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> =
flowOf(sessions.values.sortedByDescending { it.timestamp })
+ override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
+
override suspend fun getSession(sessionId: Long) = sessions[sessionId]
override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt
index b899156740..d5688b2d5a 100644
--- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt
@@ -352,6 +352,8 @@ private class InMemoryDiscoveryDao : DiscoveryDao {
override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> =
flowOf(sessions.values.sortedByDescending { it.timestamp })
+ override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
+
override suspend fun getSession(sessionId: Long): DiscoverySessionEntity? = sessions[sessionId]
override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt
index ae5dc1a806..05ac0a433c 100644
--- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt
@@ -89,6 +89,8 @@ private class FakeDiscoveryDao : DiscoveryDao {
override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> =
flowOf(sessions.values.sortedByDescending { it.timestamp })
+ override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
+
override suspend fun getSession(sessionId: Long): DiscoverySessionEntity? = sessions[sessionId]
override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])
Served by rngit 1.5.2 - Generated in 0.36s